package com.duoduo.demo.rabbitmq.rpc;

import java.io.IOException;

import org.apache.log4j.Logger;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 基于rabbitMQ的RPC服务的服务提供者 消息消费者角色，被动接收消息，然后回复消息
 * @author: yzl
 * @date: 2016-10-23
 */
public class RpcServer {

	// 交换器名称
	private static final String EXCHANGE_NAME = "demo.rpc.ex";
	// client调用的消息存储queue
	private static final String QUEUE_NAME = "demo.rpc.q";
	// 服务提供方的路由key，实际使用场景会是应用的appid或者appName
	private static final String SERVER_ROUTING_KEY = "demo.rpc.k";
	private static final Logger logger = Logger.getLogger(RpcServer.class);

	/**
	 * @param args
	 * @throws @throws IOException
	 */
	public static void main(String[] args) throws Exception {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("172.16.185.233");
		factory.setPort(5672);
		factory.setVirtualHost("/");
		factory.setUsername("dev");
		factory.setPassword("dev");

		Connection connection = factory.newConnection();
		final Channel channel = connection.createChannel();

		// 定义RPC的direct的交换器
		channel.exchangeDeclare(EXCHANGE_NAME, "topic");
		// 声明接收client调动请求的队列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		// 绑定队列
		channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, SERVER_ROUTING_KEY);

		logger.info("服务启动完成，等待接收请求.....");

		// 接收client端的请求
		channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {

			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
					throws IOException {
				// 回复ack响应
				channel.basicAck(envelope.getDeliveryTag(), false);

				logger.info("收到client调用请求-----------");
				String msg = new String(body);
				logger.info("收到的消息如下:" + msg);
				String resp = null;

				// 模拟解析需要调用的方法
				if (msg.indexOf("add") != -1) {
					String parameter = msg.substring(msg.indexOf("(") + 1, msg.indexOf(")"));
					String[] args = parameter.split(",");
					resp = String.valueOf(add(Integer.valueOf(args[0]), Integer.valueOf(args[1])));
				} else {
					resp = "method is not found!!";
				}
				logger.info("需要回调的client的queueName:" + properties.getReplyTo());
				try {
					Integer time = new java.util.Random().nextInt(5000);
					logger.info("休眠" + time + "毫秒");
					// 随即休眠，模拟服务调用方法正常业务方法耗时
					Thread.sleep(time);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				// 发送给默认的direct交换器，路由键为发送方的queueName,默认交换机会把消息转发给此路由键名称的queue
				channel.basicPublish("", properties.getReplyTo(), properties, (msg + ", resp is：" + resp).getBytes());
				logger.info("回复clent完成");
			}
		});
	}

	// 模拟的server中的方法
	private static int add(int num1, int num2) {
		logger.info("call add method, para num1 is :" + num1 + ",num2 is :" + num2);
		return num1 + num2;
	}
}
